package HandshakeJoin;

import java.util.ArrayList;
import java.util.Random;
import java.util.Calendar;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.Lists;
import org.apache.s4.base.Event;
import org.apache.s4.core.ProcessingElement;
import org.apache.s4.core.Stream;

public class GenerateDataPE extends ProcessingElement
{
    private Stream<Event> downstream;
    private int sourceId = 0;
    final private Random random = new Random(43);
    private int numFields = 4;

    @Override
    protected void onCreate() {}

    public void onTime() {
        long timestamp = Calendar.getInstance().getTimeInMillis();

        DataEvent event = new DataEvent(sourceId, numFields, timestamp);
        event.setField(0, random.nextInt(256))
                .setField(1, random.nextInt(1024))
                .setField(2, random.nextInt(4096));

        downstream.put(event);
    }

    @Override
    protected void onRemove() {}

    public GenerateDataPE setDownStream(Stream<Event> dataStream) {
        this.downstream = dataStream;

        return this;
    }

    public GenerateDataPE setRandomSeed(final long seed) {
        random.setSeed(seed);

        return this;
    }

    public GenerateDataPE setSourceId(final int id) {
        this.sourceId = id;

        return this;
    }

    public GenerateDataPE setIngestInterval(final long interval) {
        setTimerInterval(interval, TimeUnit.MILLISECONDS);

        return this;
    }

    public GenerateDataPE setNumberOfFields(final int numFields) {
        this.numFields = numFields;

        return this;
    }
}